package org.demo;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.streams.StreamsConfig;

import java.util.Collection;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @author yexi
 */
public class AdminClientDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        AdminClient adminClient = AdminClient.create(props);

        // 获取当前所有消费者组的信息
        ListConsumerGroupsResult listConsumerGroupsResult = adminClient.listConsumerGroups();
        Collection<ConsumerGroupListing> consumerGroupListings = listConsumerGroupsResult.all().get();
        System.out.println("当前kafka中所有消费者组: ");
        consumerGroupListings.forEach(System.out::println);
        System.out.println("----------------------------------------");

        DescribeConsumerGroupsResult describeConsumerGroupsResult = adminClient.describeConsumerGroups(consumerGroupListings.stream().map(ConsumerGroupListing::groupId).collect(Collectors.toList()));
        // 获取消费者组的描述信息
        Map<String, KafkaFuture<ConsumerGroupDescription>> consumerGroupDescriptions = describeConsumerGroupsResult.describedGroups();
        // 获取每个消费者组的信息
        for (Map.Entry<String, KafkaFuture<ConsumerGroupDescription>> consumerGroup : consumerGroupDescriptions.entrySet()) {
            String groupId = consumerGroup.getKey();
            KafkaFuture<ConsumerGroupDescription> descriptionFuture = consumerGroup.getValue();
            ConsumerGroupDescription description = descriptionFuture.get();
            System.out.println("Consumer Group ID: " + groupId);
            System.out.println("Coordinator: " + description.coordinator());
            System.out.println("State: " + description.state());
            System.out.println("Members: ");
            for (MemberDescription memberDescription : description.members()) {
                System.out.println(memberDescription);
            }
            System.out.println("----------------------------------------");
        }


        adminClient.close();
    }

}
